package com.test.service;

import com.test.Utils.StockTradeMapper;
import com.test.entity.StockTrade;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.triggers.CountTrigger;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import redis.clients.jedis.Jedis;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.triggers.EventTimeTrigger;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;

import java.time.Duration;
import java.util.Properties;

@Slf4j
public class TopTenStock {
    public static FlinkJedisPoolConfig redisConfig = new FlinkJedisPoolConfig.Builder()
            .setHost("192.168.43.150")
            .setPort(6379)
            .build();

    private static long counter = 0;

    public static void createDataStream(StreamExecutionEnvironment env) {
        System.out.println("开始配置TopTenStock数据流...");
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.88.135:19092,192.168.88.135:29092,192.168.88.135:39092");
        properties.setProperty("group.id", "top-ten-stock");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
            "stock-trades-",
            new SimpleStringSchema(),
            properties
        );
        consumer.setStartFromEarliest();

        // 创建数据流并添加水印
        DataStream<StockTrade> tradeStream = env
                .addSource(consumer)
                .map(new StockTradeMapper())
                .assignTimestampsAndWatermarks(
                    WatermarkStrategy
                        .<StockTrade>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                        .withTimestampAssigner((event, timestamp) -> {
                            System.out.println("分配时间戳: " + event.getTime() + " 给股票: " + event.getStockCode());
                            return event.getTime();
                        })
                );

        // 当天
        tradeStream
                .map(new MapFunction<StockTrade, Tuple2<String, Double>>() {
                    @Override
                    public Tuple2<String, Double> map(StockTrade value) throws Exception {
                        double amount = value.getTradeVolume() * value.getPrice();
                        return new Tuple2<>(value.getStockCode(), amount);
                    }
                })
                .keyBy(value -> value.f0)
                // 使用更大的窗口大小，比如10分钟
                .window(TumblingEventTimeWindows.of(Time.seconds(1)))
                // 增加水印延迟容忍时间
                .allowedLateness(Time.minutes(5))
                .aggregate(new TradeAggregator())
                .addSink(new AccumulatingRedisSink(redisConfig));

        System.out.println("TopTenStock数据流配置完成");
    }

    // 自定义 Sink 函数
    public static class AccumulatingRedisSink extends RichSinkFunction<Tuple2<String, Double>> {
        private transient Jedis jedis;
        private final FlinkJedisPoolConfig config;

        public AccumulatingRedisSink(FlinkJedisPoolConfig config) {
            this.config = config;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            try {
                super.open(parameters);
                System.out.println("尝试连接Redis: " + config.getHost() + ":" + config.getPort());
                jedis = new Jedis(config.getHost(), config.getPort());
                System.out.println("Redis连接成功");
            } catch (Exception e) {
                System.err.println("Redis连接失败: " + e.getMessage());
                e.printStackTrace();
                throw e;
            }
        }

        @Override
        public void close() throws Exception {
            if (jedis != null) {
                jedis.close();
            }
            super.close();
        }

        @Override
        public void invoke(Tuple2<String, Double> value, Context context) throws Exception {
            try {
                System.out.println("开始插入数据库...");

                // 检查Redis连接
                if (jedis == null || !jedis.isConnected()) {
                    System.out.println("Redis连接已断开，尝试重新连接...");
                    jedis = new Jedis(config.getHost(), config.getPort());
                }

                String hashKey = "stock_trade_volumes";
                String field = value.f0;

                // 打印要插入的数据
                System.out.println("正在插入数据: key=" + hashKey + ", field=" + field + ", value=" + value.f1);

                // 检查键的类型
                String type = jedis.type(hashKey);
                System.out.println("键类型: " + type);

                if (!"zset".equals(type)) {
                    System.out.println("创建新的有序集合");
                    jedis.del(hashKey);
                    jedis.zadd(hashKey, 0.0, field);
                }

                Double existingVolume = jedis.zscore(hashKey, field);
                System.out.println("现有值: " + existingVolume);

                double updatedVolume = (existingVolume == null ? 0.0 : existingVolume) + value.f1;
                System.out.println("更新后的值: " + updatedVolume);

                // 将更新后的值写回Redis
                Long result = jedis.zadd(hashKey, updatedVolume, field);
                System.out.println("插入结果: " + result);


            } catch (Exception e) {
                System.err.println("Redis操作异常: " + e.getMessage());
                e.printStackTrace();
                // 可以选择重新建立连接
                try {
                    if (jedis != null) {
                        jedis.close();
                    }
                    jedis = new Jedis(config.getHost(), config.getPort());
                } catch (Exception reconnectEx) {
                    System.err.println("重新连接失败: " + reconnectEx.getMessage());
                }
            }
        }

    }

    // 定义聚合函数
    public static class TradeAggregator implements AggregateFunction<Tuple2<String, Double>, Tuple2<String, Double>, Tuple2<String, Double>> {

        @Override
        public Tuple2<String, Double> createAccumulator() {
            System.out.println("创建累加器");
            return new Tuple2<>("", 0.0);
        }

        @Override
        public Tuple2<String, Double> add(Tuple2<String, Double> value, Tuple2<String, Double> accumulator) {
            String stockCode = value.f0;
            double totalAmount = value.f1 + accumulator.f1;
            System.out.println("Add - 股票代码: " + stockCode + ", 当前值: " + value.f1 + ", 累计值: " + totalAmount);
            return new Tuple2<>(stockCode, totalAmount);
        }

        @Override
        public Tuple2<String, Double> getResult(Tuple2<String, Double> accumulator) {
            System.out.println("GetResult - 股票代码: " + accumulator.f0 + ", 最终累计值: " + accumulator.f1);
            return accumulator;
        }

        @Override
        public Tuple2<String, Double> merge(Tuple2<String, Double> a, Tuple2<String, Double> b) {
            String stockCode = a.f0;
            double totalAmount = a.f1 + b.f1;
            System.out.println("Merge - 股票代码: " + stockCode + ", 合并值: " + totalAmount);
            return new Tuple2<>(stockCode, totalAmount);
        }
    }
}
